package pub.tbc.config.remote;

import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.zookeeper.data.Stat;
import pub.tbc.config.ConfigProvider;
import pub.tbc.config.cache.ConfigCache;
import pub.tbc.config.helper.ZkHelper;
import pub.tbc.config.service.RemoteConfigService;
import pub.tbc.config.util.Objs;

import java.util.*;

/**
 * 基于zookeeper做配置中心的远程配置实现
 *
 * @author tbc on 2017/8/15 09:50:44.
 */
@Slf4j
public class RemoteConfigServiceByZookeeper implements RemoteConfigService {
    private String ZK_IP = "";
    private CuratorFramework client = ZkHelper.getCuratorFrameworkWithOptions();
    String configPath = ResourceBundle.getBundle("application").getString("config.path.root");

    @Override
    public Map<String, String> getConfigList() {

        log.debug("config root path : {}", configPath);

        if (!isExist(configPath))
            return null;
        log.debug("config node exist...");

        List<String> childrenNodes = getChildren(configPath);
        log.debug("children node list : {}", childrenNodes);

        if (Objs.isEmpty(childrenNodes))
            return null;
        Map<String, String> data = getAndMapData(configPath, childrenNodes);
        log.debug("config data : {}", data);

        // 注册监听器
        listenChildren(configPath);
        data.keySet().stream().map(p -> configPath + "/" + p).forEach(this::listenData);
        return data;
    }

    @Override
    public void listen(ConfigCache cache, Map<String, String> data) {
        listenChildren(configPath);
        data.keySet().stream().map(p -> configPath + "/" + p).forEach(this::listenData);
    }

    private boolean isExist(String path) {
        try {
            Stat stat = client.checkExists().forPath(path);
            return stat != null;
        } catch (Exception e) {
            log.error("异常，无法判断节点是否存在: {}, e-type: ", e.getMessage(), e.getClass());
            throw new RuntimeException("判断节点是否存在失败");
        }
    }

    private List<String> getChildren(String path) {
        try {
            return client.getChildren().forPath(path);
        } catch (Exception e) {
            log.error("获取子节点列表失败: {}, e-type: ", e.getMessage(), e.getClass());
            throw new RuntimeException("获取远程配置列表失败");
        }
    }

    private Map<String, String> getAndMapData(String rootPath, List<String> paths) {
        Map<String, String> data = Maps.newHashMap();
        paths.forEach(path -> data.put(path, new String(getDate(rootPath + "/" + path))));
        return data;
    }

    private byte[] getDate(String path) {
        try {
            return client.getData().forPath(path);
        } catch (Exception e) {
            log.error("获取节点数据失败: {}, e-type: ", e.getMessage(), e.getClass());
            throw new RuntimeException("获取远程配置节点数据失败");
        }
    }

    /**
     * 子节点变化
     */
    private void listenChildren(String path) {
        log.info("registry listener for CHILDREN");
        PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                log.debug("开始进行事件分析:-----");
                ChildData data = event.getData();
                switch (event.getType()) {
                    case CHILD_ADDED:
                        log.debug("CHILD_ADDED : " + data.getPath() + "  数据:" + data.getData());
                        break;
                    case CHILD_REMOVED:
                        log.debug("CHILD_REMOVED : " + data.getPath() + "  数据:" + data.getData());
                        break;
                    case CHILD_UPDATED:
                        log.debug("CHILD_UPDATED : " + data.getPath() + "  数据:" + data.getData());
                        break;
                    default:
                        break;
                }
            }
        };
    }

    /**
     * 监听数据变化
     */
    private void listenData(String path) {
        log.info("registry listener for set data");
        //设置节点的cache
        final NodeCache nodeCache = new NodeCache(client, path, false);
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                String newData = new String(nodeCache.getCurrentData().getData());
                log.debug("the node is change and result new data: {}", newData);
                ConfigCache.instance().flushAttribute(path.substring(path.lastIndexOf("/") + 1), newData);
                log.debug("+++++>>>new data : {}", ConfigProvider.getConfig().get(path.substring(path.lastIndexOf("/") + 1)));
            }
        });
        try {
            nodeCache.start();
        } catch (Exception e) {
            log.error("node cache start failure");
            e.printStackTrace();
        }
    }
}
